package cn.com.nes.common.util;


import cn.com.nes.common.packet.AbstractPacket;
import cn.com.nes.common.packet.PacketType;
import io.netty.channel.Channel;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;

public class Transmitter {
	
	private final Channel channel;
	
	private final Map<Long, Promise.Inform<AbstractPacket>> requests = new ConcurrentHashMap<>();
	private final List<Consumer<AbstractPacket>> receiveListeners = new CopyOnWriteArrayList<>();
	
	public Transmitter(Channel channel) {
		this.channel = channel;
	}
	
	public Promise<AbstractPacket> sendAndReceiveData(AbstractPacket packet) {
		long seq = packet.getSequenceNumber();
		Promise<AbstractPacket> promise = Promise.newPromise(inform -> {
			requests.put(seq, inform);
			this.channel.writeAndFlush(packet);
		}, () -> requests.remove(seq));
		return promise;
	}
	
	public void inbound(AbstractPacket packet) {
		if(packet.getType() == PacketType.RESPONSE) {
			long seq = packet.getSequenceNumber();
			Promise.Inform<AbstractPacket> inform = requests.remove(seq);
			if(inform != null) {
				inform.resolve(packet);
			} else {
				for(Consumer<AbstractPacket> consumer : receiveListeners) {
					consumer.accept(packet);
				}
			}
		} else {
			for(Consumer<AbstractPacket> consumer : receiveListeners) {
				consumer.accept(packet);
			}
		}
	}
	
	public void shutdown() {
		Iterator<Promise.Inform<AbstractPacket>> it = requests.values().iterator();
		while(it.hasNext()) {
			it.next().reject(new InterruptedException("Transmitter shutdown"));
			it.remove();
		}
	}
	
	public Channel getChannel() {
		return channel;
	}

	public void addReceiveListener(Consumer<AbstractPacket> listener) {
		this.receiveListeners.add(listener);
	}

	public void removeReceiveListener(Consumer<AbstractPacket> listener) {
		this.receiveListeners.remove(listener);
	}
	
}
